package com.uit3j.cloud.search.listener;

import com.uit3j.cloud.api.product.bo.EsHotSearchBO;
import com.uit3j.cloud.api.product.feign.HotSearchFeignClient;
import com.uit3j.cloud.common.exception.Uit3jcloudException;
import com.uit3j.cloud.common.response.ServerResponseEntity;
import com.uit3j.cloud.common.rocketmq.config.RocketMqConstant;
import com.uit3j.cloud.common.util.Json;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.elasticsearch.action.support.WriteRequest;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.xcontent.XContentType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;

/**
 * 热搜的监听
 * @author tan
 * @date 2024/07/10
 */
@Component
@RocketMQMessageListener(topic = RocketMqConstant.HOT_SEARCH_UPDATE,consumerGroup = RocketMqConstant.HOT_SEARCH_UPDATE)
public class HotSearchUpdateConsumer implements RocketMQListener<EsHotSearchBO> {

    private static final Logger log = LoggerFactory.getLogger(HotSearchUpdateConsumer.class);

    @Autowired
    private HotSearchFeignClient hotSearchFeignClient;

    @Autowired
    private RestHighLevelClient restHighLevelClient;

    /**
     * 插入热搜时，此时插入es
     */
    @Override
    public void onMessage(EsHotSearchBO sendHotSearch) {
        Long hotSearchId = sendHotSearch.getHotSearchId();
        ServerResponseEntity<EsHotSearchBO> esHotSearchResponse = hotSearchFeignClient.loadEsHotSearchBO(hotSearchId);
        UpdateRequest request = new UpdateRequest("hot_search", String.valueOf(hotSearchId));
        request.doc(Json.toJsonString(esHotSearchResponse), XContentType.JSON);
        request.docAsUpsert(true);
        request.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
        log.info("修改热搜保存es回调...: " + sendHotSearch.toString());
        try {
            UpdateResponse updateResponse = restHighLevelClient.update(request, RequestOptions.DEFAULT);
            log.info(updateResponse.toString());
        } catch (IOException e) {
            log.error(e.toString());
            throw new Uit3jcloudException("更新订单es信息异常",e);
        }
    }


}
